package com.example.score.mq.producer;


import com.alibaba.fastjson.JSON;
import com.example.score.dto.ErrorQuestionKafkaMessageDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;

@Component
@Slf4j
public class ErrorQuestionProducer {


    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${spring.kafka.consumer.topic}")
    private String topic;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 错题数据发送到kafka
     *
     * @param errorQuestionDTO 消息体
     */
    public void setToKafka(ErrorQuestionKafkaMessageDTO errorQuestionDTO) {
        Long examId = errorQuestionDTO.getExamId();
        kafkaTemplate.send(topic, JSON.toJSONString(errorQuestionDTO)).addCallback(result -> {
            log.info("消息发送成功,examId:{},偏移量：{}:", examId, ObjectUtils.isEmpty(result) ? null : result.getRecordMetadata().offset());
        }, ex -> log.error("消息发送失败,examId:{},异常信息:{}", examId, ex.getMessage()));
    }
}
